package com.czk.rdd.practice

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author:ChenZhangKun
 * @Date: 2021/12/10 11:34
 */
object Req_01 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    // 获取原始数据
    val rdd = sc.textFile("D:\\java\\spark\\spark\\resource\\data\\agent.log")
    // 原始数据结构转换
    val mapRDD: RDD[((String, String), Int)] = rdd.map(
      // 一行行的处理
      line => {
        val datas = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    )
    // 转换后的结构进行分组转换
    // 省份广告为一个整体
    val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(_ + _)
    // 将聚合的结果进行转换
    val newMapRDD = reduceRDD.map {
      // 进行模式匹配
      case ((prv, ad), sum) => {
        (prv, (ad, sum))
      }
    }
    // 将转换结构后的数据进行分组
    // 按省份分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()
    // 将分组后的数据进行组内排序，降序取前三
    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        // 根据第二个元素排序
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )

    // 采集数据
    resultRDD.collect().foreach(println)
  }
}
